package com.example.mymessagequeue.mqserver.core;

import com.example.mymessagequeue.common.Consumer;
import com.example.mymessagequeue.common.ConsumerEnv;
import com.example.mymessagequeue.common.MqException;
import com.example.mymessagequeue.mqserver.VirtualHost;

import java.util.concurrent.*;

/**
 * 通过这个类来实现消费消息的核心逻辑
 */
public class ConsumerManager {

    // 持有上层的 VirtualHost 对象的引用, 用来操作数据
    private VirtualHost parent;

    // 指定一个线程池，负责执行回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    // 存放令牌的队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();

    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;

        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String queueName = tokenQueue.take();
                    // 2. 根据令牌找到队列
                    MsgQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManger] 取令牌后, 发现该队列名不存在! queueName=" + queueName);
                    }
                    // 3. 从这个队列中消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        // 把线程设为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    /**
     * 通知可以消费消息了，这个方法的调用时机就是发送消息的时候.
     * @param queueName
     * @throws InterruptedException
     */
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    /**
     * 添加一个消费者
     *
     * @param consumerTag
     * @param queueName
     * @param autoAck
     * @param consumer
     * @throws MqException
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MsgQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            parent.getMemoryDataCenter().consumerOnline(consumerEnv); // 绑定消费者与队列的关系
            // 如果当前队列中已经有了一些消息了, 就需要立即消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 这个方法调用一次就消费一条消息, 调用这个消息前, 先保证消费者是在线的情况下
                if (parent.getMemoryDataCenter().consumerIsAlive(consumerTag)) {
                    consumeMessage(queue);
                } else {
                    break;
                }
            }
        }
    }


    /**
     * 消费消息核心代码
     *
     * @param queue
     */
    private void consumeMessage(MsgQueue queue) throws MqException {
        // 1. 按照轮询的方式, 找个消费者出来
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        if (consumerEnv == null) {
            // 当前队列没有消费者, 暂时不消费
            return;
        }

        // 2. 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 没消息也不用消费
            return;
        }

        // 3. 把消息带入到消费者的回调方法中, 让线程池去执行
        workerPool.submit(() -> {
            try {
                // 1) 先把消息放到待确认的集合中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);

                // 2) 执行回调逻辑, 一旦回调失败, 说明消费者下线了, 需要移除当前消费者
                try {
                    consumerEnv.getConsumer()
                            .handleDelivery(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
                } catch (Exception e) {
                    // 消费者下线,
                    parent.getMemoryDataCenter().consumerOffline(consumerEnv.getConsumerTag());
                    // 此时这个消息未推送过去, 此时有两个选择， ① 直接丢弃这个消息, ② 重新将消息发送到队列， 此处使用的是方法 ②
                    String routingKey = message.getRoutingKey();
                    BasicProperties basicProperties = new BasicProperties();
                    basicProperties.setRoutingKey(basicProperties.getRoutingKey());
                    basicProperties.setDeliverMode(basicProperties.getDeliverMode());
                    byte[] body = message.getBody();
                    Message ms = Message.createMessageWithId(routingKey, basicProperties, body);

                    parent.sendMessage(queue, ms);
                    System.out.println("[ConsumerManager]消费者消费消息失败, 消费者已下线! 失败消息的 messageId: " + message.getMessageId()
                            + ", 新消息已经被重新投入到队列: new messageId = " + ms.getMessageId());
                }

                // 3) 如果当前是自动应答, 这里的程序可以直接把消息删除了
                //    如果是手动应答, 则这里程序不处理, 交给后续消费者调用 basicAck 方法
                if (consumerEnv.isAutoAck()) {
                    // 删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                        parent.executeGc(queue);
                    }

                    // 删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 删除内存中消息中心的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

    }
}
